package com.yangchao.webflux.flux;

import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FluxDemoTest {

    static Logger logger=org.slf4j.LoggerFactory.getLogger(FluxDemoTest.class);

    @DisplayName("发布者订阅者")
    @Test
    public void test01() throws InterruptedException {

        //发布数据
        // Flow.Publisher<String> publisher = new Flow.Publisher<>() {
        //     @Override
        //     public void subscribe(Flow.Subscriber<? super String> subscriber) {
        //
        //     }
        // };

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            //订阅关系
            private Flow.Subscription subscription;

            @Override //在订阅时  onXxxx：在xxx事件发生时，执行这个回调
            public void onSubscribe(Flow.Subscription subscription) {
                logger.info("订阅开始了：{}", subscription);
                this.subscription = subscription;
                //从上游请求一个数据
                subscription.request(7);
            }

            @Override //在下一个元素到达时； 执行这个回调；   接受到新数据
            public void onNext(String item) {
                logger.info("订阅者，接受到数据：{}", item);
                if (item.equals("hello7")) {
                    logger.info("订阅者 取消订阅");
                    subscription.cancel(); //取消订阅
                } else {
                    subscription.request(1);
                }
            }

            @Override //在错误发生时，
            public void onError(Throwable throwable) {
                logger.info( "订阅者，接受到错误信号：{}" , throwable.getMessage());
            }

            @Override //在完成时
            public void onComplete() {
                logger.info("订阅者，接受到完成信号：");
            }
        };

        publisher.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            if(i == 7){
                publisher.closeExceptionally(new RuntimeException("发生异常了"));
            }
            logger.info("发送数据：{}", "hello" + i);
            if(publisher.isClosed()){
                logger.info("publisher is closed");
                break;
            }
            publisher.submit("hello" + i); //buffer
        }
        logger.info("main end");
        publisher.close();
        Thread.sleep(1000L);
    }

    class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {

        private Flow.Subscription subscription; // 保存绑定关系

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            logger.info("processor订阅绑定完成");
            this.subscription = subscription;
            subscription.request(1); // 找上游要一个数据
        }

        @Override // 数据到达，触发这个回调
        public void onNext(String item) {
            logger.info("processor拿到数据：" + item);
            if(item.equals("hello6")){
                logger.info("processor 取消订阅");
                subscription.cancel(); // 取消订阅
            }
            submit(item);// 把我加工后的数据发出去
            subscription.request(1); // 再要新数据
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {

        }
    }

    @DisplayName("发布者 处理器 订阅者")
    @Test
    public void test02() throws InterruptedException {
        //发布数据
        // Flow.Publisher<String> publisher = new Flow.Publisher<>() {
        //     @Override
        //     public void subscribe(Flow.Subscriber<? super String> subscriber) {
        //
        //     }
        // };

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        MyProcessor myProcessor = new MyProcessor();
        publisher.subscribe(myProcessor);
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            //订阅关系
            private Flow.Subscription subscription;

            @Override //在订阅时  onXxxx：在xxx事件发生时，执行这个回调
            public void onSubscribe(Flow.Subscription subscription) {
                logger.info("订阅开始了：{}", subscription);
                this.subscription = subscription;
                //从上游请求一个数据
                subscription.request(7);
            }

            @Override //在下一个元素到达时； 执行这个回调；   接受到新数据
            public void onNext(String item) {
                logger.info("订阅者，接受到数据：{}", item);
                if (item.equals("hello7")) {
                    logger.info("订阅者 取消订阅");
                    subscription.cancel(); //取消订阅
                } else {
                    subscription.request(1);
                }
            }

            @Override //在错误发生时，
            public void onError(Throwable throwable) {
                logger.info( "订阅者，接受到错误信号：{}" , throwable.getMessage());
            }

            @Override //在完成时
            public void onComplete() {
                logger.info("订阅者，接受到完成信号：");
            }
        };

        myProcessor.subscribe(subscriber);
        for (int i = 0; i < 10; i++) {
            if(i == 7){
                publisher.closeExceptionally(new RuntimeException("发生异常了"));
            }
            if(publisher.isClosed()){
                logger.info("publisher is closed");
                break;
            }
            logger.info("发送数据：{}", "hello" + i);
            publisher.submit("hello" + i); //buffer
        }
        logger.info("main end");
        publisher.close();
        Thread.sleep(1000L);
    }

}
